"""
Case Type   : 协议兼容
Case Name   : sha2协议连接成功，创建及修改分区表，部分测试点合理报错
Create At   : 2024.09.23
Owner       : @zhanghuan96
Description :
    1、创建兼容b库并查询插件是否加载成功
    2、创建用户
    3、caching_sha2认证
    4、查询系统未使用端口
    5、设置协议兼容guc参数
    6、连接og侧B库创建分区表及索引
    7、og数据库查询分区信息
    8、drop partition
    9、删除全部分区
    10、删除分区
    11、删除最后一个分区
    12、删除不存在的分区
    13、og数据库查询分区信息
    14、清理环境
Expect      :
    1、成功，插件查询成功
    2、成功
    3、成功
    4、成功
    5、成功
    6、成功
    7、成功
    8、成功
    9、报错
    10、报错
    11、报错
    12、报错
    13、成功
    14、成功
History     :
"""
import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonMySQL import CommonMySQL
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class CompatibilityTest01(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.sh_pri = CommonSH('PrimaryDbUser')
        self.pri_node = Node('PrimaryDbUser')
        self.my_node = Node('mysql8')
        self.constant = Constant()
        self.db_name = 'db_protocol_0016'
        self.u_name = 'u_protocol_0016'
        self.tb_name = 'tb_protocol_0016'
        self.inx_name = 'index_protocol_0016'
        self.conf_path = os.path.join(macro.DB_INSTANCE_PATH,
                                      'postgresql.conf')
        self.default_value1 = self.common.show_param('enable_dolphin_proto')
        self.default_value2 = self.common.show_param('dolphin_server_port')
        self.exec_sql1 = f'''drop table if exists {self.tb_name};
        create table {self.tb_name}(c1 int primary key,c2 timestamp)\
        partition by range(c2) interval('1 day')\
        (partition p1 values less than ('1990-01-01 00:00:00'),\
        partition p2 values less than ('1990-01-02 00:00:00'),\
        partition p3 values less than ('1990-01-03 00:00:00'),\
        partition p4 values less than ('1990-01-04 00:00:00'),\
        partition p5 values less than ('1990-01-05 00:00:00'),\
        partition p6 values less than ('1990-01-06 00:00:00'),\
        partition p7 values less than ('1990-01-07 00:00:00'),\
        partition p8 values less than ('1990-01-08 00:00:00'),\
        partition p9 values less than ('1990-01-09 00:00:00'),\
        partition p10 values less than ('1990-01-10 00:00:00')) ;
        drop index if exists {self.inx_name};
        create index {self.inx_name} on {self.tb_name} (c1) global;
        drop index if exists {self.inx_name}_01;
        create index {self.inx_name}_01 on {self.tb_name} (c2) local;
        insert into {self.tb_name} values (1,'1990-01-01 00:00:00'),\
         (2,'1990-01-02 00:00:00'),(3,'1990-01-03 00:00:00'),\
         (4,'1990-01-04 00:00:00'),(5,'1990-01-05 00:00:00');'''
        self.exec_sql2 = f'''alter table {self.tb_name} drop partition p1;
         alter table {self.tb_name} drop partition p2,p3;'''
        self.exec_sql3 = f'''alter table {self.tb_name} drop partition p4,p5,\
        p6,p7,p8,p9,p10;'''
        self.exec_sql4 = f'''alter table {self.tb_name} drop partition p4,p5,\
        p6,p7,p8,p9;'''
        self.exec_sql5 = f'''alter table {self.tb_name} drop partition p10;'''
        self.exec_sql6 = f'''alter table {self.tb_name} drop partition p1;'''

    def test_create_function(self):
        text = '-----step1:创建兼容b库并查询插件;expect:成功-----'
        self.log.info(text)
        sql_cmd = (f"drop database if exists {self.db_name};"
                   f"create database {self.db_name} encoding 'utf8' "
                   f"dbcompatibility ='B';")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertEqual(1,
                         sql_res.count(self.constant.CREATE_DATABASE_SUCCESS),
                         '兼容B库创建失败')
        sql_cmd1 = "select extname from pg_extension where extname ='dolphin';"
        self.log.info(sql_cmd1)
        sql_res1 = self.sh_pri.execut_db_sql(sql_cmd1, dbname=self.db_name)
        self.log.info(sql_res1)
        self.assertIn('dolphin', sql_res1, '查询失败')

        text = '---step2:创建用户;expect:成功---'
        self.log.info(text)
        sql_cmd2 = (f"drop user if exists {self.u_name};"
                    f"create user {self.u_name} with password "
                    f"'{macro.COMMON_PASSWD}';"
                    f"grant all privileges to {self.u_name};")
        self.log.info(sql_cmd2)
        sql_res2 = self.sh_pri.execut_db_sql(sql_cmd2, dbname=self.db_name)
        self.log.info(sql_res2)
        self.assertIn('CREATE ROLE', sql_res2, '创建失败')
        self.assertIn('ALTER ROLE', sql_res2, '创建失败')

        text = '---step3:caching_sha2认证;expect:成功---'
        self.log.info(text)
        sql_cmd = (f"select set_caching_sha2_password('{self.u_name}', "
                   f"'{macro.COMMON_PASSWD}', '');")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('(1 row)', sql_res, '创建失败')

        text = '----step4:查询系统未使用端口;expect:成功----'
        self.log.info(text)
        port = self.common.get_not_used_port(self.pri_node)
        self.assertNotEqual(0, port, '执行失败:' + text)

        text = '-step5:设置协议兼容guc参数;expect:成功-'
        self.log.info(text)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'enable_dolphin_proto = on')
        self.log.info(msg)
        self.assertTrue(msg)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'dolphin_server_port ={port}')
        self.log.info(msg)
        self.assertTrue(msg)
        sed_cmd = (f"sed -i '$a dolphin.default_database_name={self.db_name}' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '-step6:连接og侧B库创建分区表及索引;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql1, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step7:og数据库查询分区信息;expect:成功--'
        self.log.info(text)
        sql_cmd = (f"use {self.u_name};"
                   f"select relname,boundaries from pg_partition where "
                   f"parentid in (select oid from pg_class where relname "
                   f"= '{self.tb_name}') order by relname;")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('(11 rows)', sql_res)
        lines = [line.strip() for line in sql_res.strip().split('\n')]
        expected_relnames = [
            'p1', 'p10', 'p2', 'p3', 'p4', 'p5', 'p6', 'p7', 'p8', 'p9',
            'tb_protocol_0016']
        for i, relname in enumerate(expected_relnames, start=3):
            assert relname in lines[i], f"Expected '{relname}' in line {i}, \
                            but found '{lines[i]}'"

        text = '-step8:drop partition;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql2, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step9:删除全部分区;expect:报错--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql3, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('Cannot drop the only partition of a '
                      'partitioned table', c_res)

        text = '-step10:删除分区;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql4, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step11:删除最后一个分区;expect:报错--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql5, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('Cannot drop the only partition of a partitioned '
                      'table', c_res)

        text = '-step12:删除不存在的分区;expect:报错--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql6, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('partition "p1" does not exist', c_res)

        text = '-step13:og数据库查询分区信息;expect:成功--'
        self.log.info(text)
        sql_cmd = (f"use {self.u_name};"
                   f"select relname,boundaries from pg_partition where "
                   f"parentid in (select oid from pg_class where relname "
                   f"= '{self.tb_name}') order by relname;")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn(' p10              | {"1990-01-10 00:00:00+08"}',
                      sql_res)
        self.assertIn('tb_protocol_0016 |', sql_res)
        self.assertIn('(2 rows)', sql_res)

    def tearDown(self):
        text = '-----step14:清理环境;expect:成功-----'
        self.log.info(text)
        drop_cmd = f"drop database if exists {self.db_name};"
        self.log.info(drop_cmd)
        drop_res = self.sh_pri.execut_db_sql(drop_cmd, dbname='postgres')
        self.log.info(drop_res)
        drop_cmd1 = f"drop user if exists {self.u_name};"
        self.log.info(drop_cmd1)
        drop_res1 = self.sh_pri.execut_db_sql(drop_cmd1)
        self.log.info(drop_res1)

        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'enable_dolphin_proto = '
                                            f'{self.default_value1}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)
        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'dolphin_server_port ='
                                            f'{self.default_value2}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)

        sed_cmd = (f"sed -i '/dolphin.default_database_name={self.db_name}/d' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.log.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)
        self.assertIn('DROP ROLE', drop_res1, '清理环境失败')
        self.assertEqual(1,
                         drop_res.count(self.constant.DROP_DATABASE_SUCCESS),
                         '清理环境失败')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
